package current.queues;

import java.util.concurrent.LinkedTransferQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 模拟线程池的生成消费
 */
public class LinkedTransferQueueDemo {

    static class Scheduler {
        LinkedTransferQueue<Runnable> tasks = new LinkedTransferQueue<>();

        static AtomicInteger idCount = new AtomicInteger(0);

        class Worker implements Runnable {
            int id;
            public Worker() {
                this.id = idCount.getAndIncrement();
            }

            @Override
            public void run() {
                while (true) {
                    Runnable runnable = null;
                    try {
                        runnable = tasks.take();
                        runnable.run();
                        System.out.format("work done by id = %d\n", id);
                        TimeUnit.MILLISECONDS.sleep(100);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }

        public Scheduler(int workers) {
            for (int i = 0; i < workers; i++) {
                new Thread(new Worker()).start();
            }
        }

        public void submit(Runnable r) {
            // tryTransfer有双向,反向压力能力，比起普通的offer线程数可以开到更高
            // 而普通的offer只能固定10个，tryTransfer会等到有足够多的消费者
            // 可以更好的、弹性的利用资源，而不是写死的固定资源
            while (!tasks.tryTransfer(r)) {
                // onSpinWait只会wait少量的CPU周期
                Thread.onSpinWait();

                // tryTransfer失败就重新创建
                new Thread(new Worker()).start();
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        Scheduler scheduler = new Scheduler(10);

        for (int i = 0; i < 1000; i++) {
            int localI = i;
            TimeUnit.MILLISECONDS.sleep(1);
            scheduler.submit(() -> {
                try {
                    TimeUnit.MILLISECONDS.sleep(1);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
    }
}
